use crate::config::app_error::{AppError, AppResult};
use crate::extension::index::index::query::index_attribute::IndexAttribute;
use crate::extension::index::index::query::index_spec::IndexSpec;
use halo_model::{ExtensionOperator, GVK};
use serde_json::Value;
use std::any::{Any, TypeId};
use std::collections::HashMap;
use std::hash::Hash;
use std::sync::RwLock;

/// 索引规范接口

pub trait IndexSpecs: Send + Sync {
    /// 添加索引规范（类型擦除版本）
    fn add<T: ExtensionOperator + GVK>(
        &self,
        index_spec: IndexSpec<T>,
    ) -> AppResult<()>;

    /// 获取所有索引规范的名称
    fn get_index_specs<T: ExtensionOperator + GVK>(&self) -> AppResult<Vec<IndexSpec<T>>>;

    /// 获取特定名称的索引规范（类型擦除版本）
    fn get_index_spec<T: ExtensionOperator + GVK>(
        &self,
        index_name: &String,
    ) -> AppResult<Option<IndexSpec<T>>>;

    /// 检查是否包含指定名称的索引规范
    fn contains(&self, index_name: &String) -> bool;

    /// 移除指定名称的索引规范
    fn remove(&self, index_name: &String) -> AppResult<()>;
}

/// 类型擦除的默认索引规范实现
///
/// 这个结构体使用类型擦除技术，可以存储不同泛型类型的 IndexSpec
pub struct DefaultIndexSpecs {
    // 使用类型擦除存储不同类型的 IndexSpec
    // Key: index_name, Value: (TypeId, Box<dyn Any + Send + Sync>)
    index_specs: RwLock<HashMap<String, (TypeId, Box<dyn Any + Send + Sync>)>>,
}

impl DefaultIndexSpecs {
    pub fn new() -> Self {
        DefaultIndexSpecs {
            index_specs: RwLock::new(HashMap::new()),
        }
    }

    fn check_index_spec<T: ExtensionOperator + GVK>(spec: &IndexSpec<T>) -> AppResult<()> {
        if spec.name.is_empty() {
            return Err(anyhow::anyhow!("IndexSpec name must not be blank").into());
        }
        Ok(())
    }
}

impl IndexSpecs for DefaultIndexSpecs {
    fn add<T: ExtensionOperator + GVK>(
        &self,
        index_spec: IndexSpec<T>,
    ) -> AppResult<()> {
        Self::check_index_spec(&index_spec)?;

        let index_name = index_spec.name.clone();
        let type_id = TypeId::of::<IndexSpec<T>>();

        let mut specs = self.index_specs.write().map_err(|_| {
            AppError::InternalServerError("Failed to acquire write lock on index_specs".to_string())
        })?;

        if specs.contains_key(&index_name) {
            return Err(
                anyhow::anyhow!("IndexSpec with name {} already exists", index_name).into(),
            );
        }

        specs.insert(index_name, (type_id, Box::new(index_spec)));
        Ok(())
    }

    fn get_index_specs<T: ExtensionOperator + GVK>(&self) -> AppResult<Vec<IndexSpec<T>>> {
        let specs = self.index_specs.read().map_err(|_| {
            AppError::InternalServerError("Failed to acquire read lock on index_specs".to_string())
        })?;

        let mut vec = Vec::new();
        for (type_id, v) in specs.iter() {
            if let Some(spec) = v.downcast_ref::<IndexSpec<T>>() {
                vec.push(spec.clone());
            }
        }
        Ok(vec)
    }

    fn get_index_spec<T: ExtensionOperator + GVK>(
        &self,
        index_name: &String,
    ) -> AppResult<Option<IndexSpec<T>>> {
        let specs = self.index_specs.read().map_err(|_| {
            AppError::InternalServerError("Failed to acquire read lock on index_specs".to_string())
        })?;

        if let Some((type_id, boxed_spec)) = specs.get(index_name) {
            let expected_type_id = TypeId::of::<IndexSpec<T>>();
            if *type_id == expected_type_id {
                if let Some(spec) = boxed_spec.downcast_ref::<IndexSpec<T>>() {
                    Ok(Some(spec.clone()))
                }
            }
        }
        Ok(None)
    }

    fn contains(&self, index_name: &String) -> bool {
        if let Ok(specs) = self.index_specs.read() {
            specs.contains_key(index_name)
        } else {
            false
        }
    }

    fn remove(&self, index_name: &String) -> AppResult<()> {
        let mut specs = self.index_specs.write().map_err(|_| {
            AppError::InternalServerError("Failed to acquire write lock on index_specs".to_string())
        })?;
        specs.remove(index_name);
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::extension::index::index::index_specs::{DefaultIndexSpecs, IndexSpecs};
    use crate::tests::primary_key_index_spec;

    #[test]
    fn test_type_erased_add() {
        let specs = DefaultIndexSpecs::new();
        specs.add(primary_key_index_spec()).unwrap();
        assert!(specs.contains(&"metadata.name".to_string()));
    }

    #[test]
    fn test_type_erased_add_with_exception() {
        let specs = DefaultIndexSpecs::new();
        let name_spec = primary_key_index_spec();
        specs.add(name_spec).unwrap();
        let name_spec = primary_key_index_spec();
        let result = specs.add(name_spec);
        assert!(result.is_err());
        match result {
            Err(e) => assert!(e.to_string().contains("already exists")),
            _ => panic!("should be error"),
        }
    }

    #[test]
    fn test_type_erased_get_all() {
        let specs = DefaultIndexSpecs::new();
        specs.add(primary_key_index_spec()).unwrap();
        let index_specs = specs.get_index_specs().unwrap();
        assert_eq!(1, index_specs.len());
    }

    #[test]
    fn test_type_erased_get_specific() {
        let specs = DefaultIndexSpecs::new();
        specs.add(primary_key_index_spec()).unwrap();
        let index_spec = specs.get_index_spec(&"metadata.name".to_string()).unwrap();
        assert!(index_spec.is_some());
        let index_spec = index_spec.unwrap();

        // let any = index_spec.as_any();
        // assert!(any.is::<IndexSpec<FunctionalIndexAttribute<FakeExtension>>>());

        // let b = any
        //     .downcast_ref::<IndexSpec<FunctionalIndexAttribute<FakeExtension>>>()
        //     .unwrap();
        // let a = primary_key_index_spec();
        // assert_eq!(a, *b);
    }

    #[test]
    fn test_type_erased_remove() {
        let specs = DefaultIndexSpecs::new();
        specs.add(primary_key_index_spec()).unwrap();
        assert!(specs.contains(&"metadata.name".to_string()));
        specs.remove(&"metadata.name".to_string()).unwrap();
        assert!(!specs.contains(&"metadata.name".to_string()));
    }
}
